package com

import (
	"strings"
	"os"
	"time"
	"github.com/bugfan/logrus"
	"github.com/bugfan/to"	
	"github.com/nsqio/go-nsq"
	"log"
	"sync"
)
var NsqLog bool
var Add int64
var NsqAddNum int64
var NsqData *NsqSync=new(NsqSync)
var NsqRestart chan bool
func init(){
	NsqLog=to.Bool(os.Getenv("NsqLog"))					//nsq 数据log
	NsqAddNum=to.Int64(os.Getenv("NsqAddNum"))			//本地控制频率
	NsqData.Data=make(map[string]string)
	NsqRestart=make(chan bool,1)
}
type NsqSync struct{
	sync.Mutex
	Data map[string]string
}
func NsqDeamon(){
	go Nsq()
	for{
		select{
		case <-NsqRestart:
			go Nsq()
		}
		time.Sleep(10e9)
	}
}
func Nsq(){
	topics:=strings.Split(os.Getenv("NsqTopics"),"^")
	channel:=os.Getenv("NsqChannel")
	if len(topics) <1 || channel==""{
		logrus.Error("NSQ:未拿到 topic或channel,无法启动,请检查环境变量")
		return 
	}
	for _,v:=range topics{
		if bindTopic(v,channel)!=0{
			NsqRestart <- true
			logrus.Error("NSQ:连接失败,请检查连接参数")			
			return 
		}
	}
	<- (chan int)(nil)
}
func bindTopic(t string,ch string)int{								
	c, err := nsq.NewConsumer(t, ch, nsq.NewConfig())
	if err != nil {
		log.Println(err)
		return -1
	}
	c.AddHandler(nsq.HandlerFunc(func(m *nsq.Message) error {
		// logrus.Printf("[%s] %s: %s\n", m.NSQDAddress, string(m.Body))
		NsqData.Lock()
		NsqData.Data[t]=string(m.Body)
		NsqData.Unlock()
		if NsqLog{
			logrus.Println("NSQ "+t+" Data:",NsqData.Data[t])	
		}
		m.Finish()
		return nil
	}))                                           // 添加消息处理
	if err := c.ConnectToNSQD(os.Getenv("NSQ_PUMP_ADDR")); err != nil {            // 建立连接
		log.Println(err)
		return -2	
	}
	return 0
}
